KAFKA-13958: Expose logdirs total and usable space via Kafka API (KIP… by mimaison · Pull Request #12248 · apache/kafka

mimaison · 2022-06-03T15:50:35Z

…-827)

This implements KIP-827: https://cwiki.apache.org/confluence/display/KAFKA/KIP-827%3A+Expose+logdirs+total+and+usable+space+via+Kafka+API

Add TotalBytes and UsableBytes to DescribeLogDirsResponse
Add matching getters on LogDirDescription

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

soarez

Thanks for this PR

core/src/main/scala/kafka/server/ReplicaManager.scala

…-827) This implements KIP-827: https://cwiki.apache.org/confluence/display/KAFKA/KIP-827%3A+Expose+logdirs+total+and+usable+space+via+Kafka+API Add TotalBytes and UsableBytes to DescribeLogDirsResponse Add matching getters on LogDirDescription

divijvaidya · 2022-06-08T11:07:18Z

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

+    assertEquals(mockLogMgr.liveLogDirs.size, responses.size)
+    responses.foreach { response =>
+      assertEquals(Errors.NONE.code, response.errorCode)
+      assertTrue(response.totalBytes >= 0)


Is there a reason due to which totalBytes could be 0? Perhaps we want to assert a > 0 condition here?

True this assertion is a bit too defensive! > 0 should be fine. I'll push an update.

divijvaidya · 2022-06-08T11:14:11Z

clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java

 public class DescribeLogDirsResponse extends AbstractResponse {

    public static final long INVALID_OFFSET_LAG = -1L;
+    public static final long UNKNOWN_VOLUME_BYTES = -1L;


Do we still want this?
Correct me if I am wrong here but I thought that we reached a conclusion in the KIP that Optional will cover the scenario when client is using a newer API and broker is old.

The wire protocol does not have nullable integers. So the new fields are defined to use -1 as the default value in the protocol:

kafka/clients/src/main/resources/common/message/DescribeLogDirsResponse.json

Lines 52 to 56 in a73b96a

{ "name": "TotalBytes", "type": "int64", "versions": "4+", "ignorable": true, "default": "-1",

"about": "The total size in bytes of the volume the log directory is in."

},

{ "name": "UsableBytes", "type": "int64", "versions": "4+", "ignorable": true, "default": "-1",

"about": "The usable size in bytes of the volume the log directory is in."

This constant is used to map the value in DescribeLogDirsResponse to an Optional:

kafka/clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

Lines 44 to 45 in a73b96a

this.totalBytes = (totalBytes == UNKNOWN_VOLUME_BYTES) ? OptionalLong.empty() : OptionalLong.of(totalBytes);

this.usableBytes = (usableBytes == UNKNOWN_VOLUME_BYTES) ? OptionalLong.empty() : OptionalLong.of(usableBytes);

divijvaidya · 2022-06-08T11:15:46Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+    private final OptionalLong usableBytes;

    public LogDirDescription(ApiException error, Map<TopicPartition, ReplicaInfo> replicaInfos) {
+        this(error, replicaInfos, UNKNOWN_VOLUME_BYTES, UNKNOWN_VOLUME_BYTES);


Instead of UNKNOWN_VOLUME_BYTES, this could be null or Optional.Empty()? That way we could get rid of UNKNOWN_VOLUME_BYTES completely.

tombentley

Thanks @mimaison

tombentley · 2022-06-08T15:00:31Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

                }
            }
-            result.put(logDirResult.logDir(), new LogDirDescription(Errors.forCode(logDirResult.errorCode()).exception(), replicaInfoMap));
+            result.put(logDirResult.logDir(), new LogDirDescription(Errors.forCode(logDirResult.errorCode()).exception(), replicaInfoMap, logDirResult.totalBytes(), logDirResult.usableBytes()));


clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

tombentley · 2022-06-08T15:02:06Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+        return totalBytes;
+    }
+
+    public OptionalLong usableBytes() {


Also do we want to say something about the contraints on usableBytes and totalBytes (individually and compared with each other).

I've added some details, let me know if it's not what you had in mind

divijvaidya · 2022-06-08T15:35:43Z

Thank you for making the suggested changes.

The current set of changes don't contain a fully end to end integration test (KafkaAdminClientTest uses response mocks). Please make a change in DescribeLogDirsRequestTest.scala to validate the new changes.

mimaison · 2022-06-10T13:09:30Z

Thanks @divijvaidya for the review. I've updated the end to end tests to cover this new feature.

divijvaidya

LGTM!

viktorsomogyi

@mimaison thanks for this feature, it's a really useful addition for us as well.
Tried out your code locally, ran the tests, they all work. I don't have any particular comments, it looks good to me.

tombentley

LGTM, thanks @mimaison!

soarez reviewed Jun 7, 2022

View reviewed changes

core/src/main/scala/kafka/server/ReplicaManager.scala Outdated Show resolved Hide resolved

mimaison added 2 commits June 7, 2022 17:02

Cleanup

a73b96a

mimaison force-pushed the kip-827 branch from 9083fe5 to a73b96a Compare June 7, 2022 15:02

divijvaidya reviewed Jun 8, 2022

View reviewed changes

mimaison added the kip Requires or implements a KIP label Jun 8, 2022

Adjust assertion in ReplicaManagerTest

ec61022

tombentley reviewed Jun 8, 2022

View reviewed changes

Address feedback (javadocs + formatting)

9af1566

Add end to end tests

df8aea2

divijvaidya approved these changes Jun 10, 2022

View reviewed changes

viktorsomogyi approved these changes Jun 10, 2022

View reviewed changes

tombentley approved these changes Jun 14, 2022

View reviewed changes

mimaison merged commit 4fcfd9d into apache:trunk Jun 14, 2022

mimaison deleted the kip-827 branch June 14, 2022 12:20

leweafan mentioned this pull request May 14, 2023

Metricbeat Kafka module - add topic metrics elastic/beats#35457

Closed

	{ "name": "TotalBytes", "type": "int64", "versions": "4+", "ignorable": true, "default": "-1",
	"about": "The total size in bytes of the volume the log directory is in."
	},
	{ "name": "UsableBytes", "type": "int64", "versions": "4+", "ignorable": true, "default": "-1",
	"about": "The usable size in bytes of the volume the log directory is in."

	this.totalBytes = (totalBytes == UNKNOWN_VOLUME_BYTES) ? OptionalLong.empty() : OptionalLong.of(totalBytes);
	this.usableBytes = (usableBytes == UNKNOWN_VOLUME_BYTES) ? OptionalLong.empty() : OptionalLong.of(usableBytes);

Conversation

mimaison commented Jun 3, 2022

Committer Checklist (excluded from commit message)

Uh oh!

soarez left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tombentley left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

divijvaidya commented Jun 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mimaison commented Jun 10, 2022

Uh oh!

divijvaidya left a comment

Choose a reason for hiding this comment

Uh oh!

viktorsomogyi left a comment

Choose a reason for hiding this comment

Uh oh!

tombentley left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

divijvaidya commented Jun 8, 2022 •

edited

Loading